// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <google/protobuf/util/json_util.h>
#include <pybind11/pybind11.h>

#include "appcoin2/support/py_module/driver/py_feed_update.h"
#include "appcoin2/support/py_module/driver/py_order_update.h"
#include "coin2/app/driver.h"
#include "coin2/exchange/base/log/strategy_logger.h"
#include "coin2/exchange/base/order/order.h"
#include "coin2/strategy/strategy.h"
#include "coin2/strategy/util/latency_util.h"

namespace coin2::exchange::driver {

namespace impl {

namespace py = pybind11;
using base::feed::FeedUpdate;
using coin2::exchange::base::order::OrderUpdate;
using coin2::exchange::base::strategy_util::StrategyLogger;
using coin2::exchange::base::strategy_util::TelemetryLogger;

class PickleForwarder : public ::coin2::strategy::IStrategy {
 public:
  PickleForwarder(py::object pipe_incoming, py::object pipe_outgoing, bool enable_telemetry)
      : pipe_incoming_{pipe_incoming},
        pipe_outgoing_{pipe_outgoing},
        enable_telemetry_{enable_telemetry} {
    CHECK(py::hasattr(pipe_incoming_, "recv_bytes"));
    CHECK(py::hasattr(pipe_outgoing_, "send_bytes"));
  }

  // feed
  void onBookFeed(const FeedUpdate& upd) override {
    PyBookViewPickleable book(
        coin2::exchange::feed::FeedType_Name(upd.feed_type()),
        upd.mea().String(),
        coin::proto::MarketType_Name(upd.mea().market),
        coin::proto::ExchangeType_Name(upd.mea().exchange),
        upd.product().relative_norm(),
        upd.product().absolute_norm(),
        upd.product().order_native_symbol(),
        upd.timestamp(),
        levels_as_list(upd.book().AskSize(), upd.book().AskBegin(), upd.book().AskEnd()),
        levels_as_list(upd.book().BidSize(), upd.book().BidBegin(), upd.book().BidEnd()));
    auto msg = py::module::import("pickle").attr("dumps")(book);
    DoSend(std::move(msg));

    if (enable_telemetry_) {
      auto acc_request = GetAccRequest(upd.mea());
      auto feed_latency_tag = GetFeedLatencyTag(upd);
      coin2::strategy::util::SetFeedTimeToLatency(
          strat_logger_.get(),
          feed_mgr_,
          upd,
          feed_latency_tag,
          acc_request,
          true);
      coin2::strategy::util::SetFeedTimeToLatency(
          strat_logger_.get(),
          feed_mgr_,
          upd,
          coin::proto::LatencyProto::ORDER_LATENCY,
          acc_request,
          false);
    }
  }

  void onTradeFeed(const FeedUpdate& upd) override {
    PyTradeViewPickleable trade(
        coin2::exchange::feed::FeedType_Name(upd.feed_type()),
        upd.mea().String(),
        coin::proto::MarketType_Name(upd.mea().market),
        coin::proto::ExchangeType_Name(upd.mea().exchange),
        upd.product().relative_norm(),
        upd.product().absolute_norm(),
        upd.product().order_native_symbol(),
        upd.timestamp(),
        upd.trade().price,
        upd.trade().fill_qty,
        upd.trade().side,
        upd.trade().has_more_trade);
    auto msg = py::module::import("pickle").attr("dumps")(trade);
    DoSend(std::move(msg));
  }

  void onLiquidationFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onOpenInterestFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onMarkPriceFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onIndexFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onFundingRateFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onIvGreeksFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onNavFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onKlineFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onWalletFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  void onMarketAnalysisFeed(const exchange::base::feed::FeedUpdate&) override {
    // TODO:
  }

  // void onHeartbeatFeed(const FeedTimer& timer) override {
  //   // TODO:
  // }

  // order
  void onAccountOrder(const OrderUpdate& upd) override {
    PyOrderUpdatePickleable order(
        coin::proto::OrderUpdateType_Name(upd.update_type()),
        upd.mea().String(),
        coin::proto::MarketType_Name(upd.mea().market),
        coin::proto::ExchangeType_Name(upd.mea().exchange),
        upd.product().relative_norm(),
        upd.product().absolute_norm(),
        upd.product().order_native_symbol(),
        upd.timestamp(),
        ProtoToDict(upd.product_order_info().event()));
    auto msg = py::module::import("pickle").attr("dumps")(order);
    DoSend(std::move(msg));
  }

  void onAccountInfo(const OrderUpdate& upd) override {
    PyAccountInfoPickleable info(
        coin::proto::OrderUpdateType_Name(upd.update_type()),
        upd.mea().String(),
        coin::proto::MarketType_Name(upd.mea().market),
        coin::proto::ExchangeType_Name(upd.mea().exchange),
        upd.account_info().owner(),
        ProtoToDict(upd.account_info().GetAccountBalance()),
        ProtoToDict(upd.account_info().GetAccountPosition()),
        upd.timestamp());
    auto msg = py::module::import("pickle").attr("dumps")(info);
    DoSend(std::move(msg));
  }

  void onOrderSubsystemError(const exchange::base::order::OrderUpdate&) override {
    // TODO:
  }

  void onOrderLog(const coin::proto::OrderGatewayLog&) override {
    // TODO:
  }

  void Init(::coin2::app::Driver* driver) override {
    if (enable_telemetry_) {
      const auto& strategy_config = driver->strategy()->config();
      coin::proto::StrategyRequestProto strat_request;
      strat_request.set_strategy_group(strategy_config.strategy_group());
      strat_request.set_strategy_name(strategy_config.strategy_name());

      const auto& strat_logger_config = strategy_config.strategy_logger_config();
      const auto& telemetry_logger_config = strategy_config.telemetry_logger_config();
      strat_logger_ = std::make_unique<StrategyLogger>(
          strat_request,
          strat_logger_config.log_root(),
          strat_logger_config.machine());
      telemetry_logger_ = std::make_unique<TelemetryLogger>(
          strat_request,
          telemetry_logger_config.log_root(),
          telemetry_logger_config.machine());
      strat_logger_->AddCallback(std::bind(
          &TelemetryLogger::CondWrite,
          telemetry_logger_.get(),
          std::placeholders::_1,
          std::placeholders::_2,
          std::placeholders::_3));

      const auto& systems = driver->order()->GetAllSystems();
      for (const auto& os : systems) {
        os->gateway()->set_latency_recorder(strat_logger_->mutable_latency_recorder());
      }
    }
  }

  void onCleanup(::coin2::app::Driver* driver) override {
    // TODO:
  }

  void InjectFeed(const coin2::feed::mgr::impl::FeedSubsystemManager* feed_mgr) {
    feed_mgr_ = feed_mgr;
  }

  void RegisterMonitor(coin2::app::Driver* driver) {
    driver->PostPeriodicTask(
        std::bind(&PickleForwarder::DoMonitor, this),
        1'000'000LL /* period */,
        1'000'000LL /* clock */);
    driver->PostPeriodicTask(
        std::bind(&PickleForwarder::DoSendOrderSnapshot, this),
        1'000'000'000LL /* period */,
        1'000'000LL /* clock */);
    driver_ = driver;
  }

 private:
  template <class T>
  bool DoSend(T&& msg) {
    pipe_outgoing_.attr("send_bytes")(std::move(msg));
    return true;
  }

  template <class T>
  bool DoRecv(T* msg) {
    auto ready = pipe_incoming_.attr("poll")();
    if (py::str(ready).is(py::str(Py_False))) {
      return false;
    }
    *msg = pipe_incoming_.attr("recv_bytes")();
    return true;
  }

  void DoMonitor() {
    py::bytes msg;
    while (DoRecv(&msg)) {
      // TODO: check exec_spec is None
      HandleExecution(msg);
    }
  }

  void DoSendOrderSnapshot() {
    auto order_list = CaptureOrderSnapshot();
    PyOrderSnapshotPickleable snapshot(order_list);
    auto msg = py::module::import("pickle").attr("dumps")(snapshot);
    DoSend(std::move(msg));
  }

  template <class T>
  bool HandleExecution(T&& msg) {
    auto obj = py::module::import("pickle").attr("loads")(msg);
    auto exec_spec = py::cast<PyExecSpecPickleable>(obj);

    switch (exec_spec.exec_type()) {
      case ExecutionType::Submit: {
        auto* os = driver_->order()->GetUniqueSystemByName(exec_spec.os_name());
        if (!os->is_ready()) {
          LOG(INFO) << "og is not ready";
          return false;
        }
        auto order_spec = GenOrderSpecFromDict(os, exec_spec.order_spec());
        if (!order_spec) {
          return false;
        }
        os->gateway()->SubmitOrder(*order_spec);
        return true;
      }
      case ExecutionType::Cancel: {
        auto* os = driver_->order()->GetUniqueSystemByName(exec_spec.os_name());
        if (!os->is_ready()) {
          LOG(INFO) << "og is not ready";
          return false;
        }
        auto proc_order_id = exec_spec.proc_order_id();
        os->gateway()->CancelOrder(proc_order_id);
        return true;
      }
      default: {
        LOG(ERROR) << "unsupported execution type: " << exec_spec.exec_type();
        return false;
      }
    }
  }

  auto CaptureOrderSnapshot() -> py::list {
    py::list ret;
    auto& systems = driver_->order()->GetAllSystems();
    for (auto& os : systems) {
      for (auto* order : os->gateway()->order_manager().GetAllWorkingOrders()) {
        ret.attr("append")(OrderContextToDict(*order));
      }
    }
    return ret;
  }

  auto GetFeedLatencyTag(const FeedUpdate& upd) -> coin::proto::LatencyProto::LatencyTag {
    if (upd.is_book())
      return coin::proto::LatencyProto::BOOK_FEED_LATENCY;
    else if (upd.is_trade())
      return coin::proto::LatencyProto::TRADE_FEED_LATENCY;
    else
      return coin::proto::LatencyProto::FEED_LATENCY;
  }

  auto GetAccRequest(const MarketExchangeApi& mea) -> coin::proto::AccountRequestProto {
    coin::proto::AccountRequestProto acc_request;
    acc_request.set_market_type(MarketType_Name(mea.market));
    acc_request.set_exchange(ExchangeType_Name(mea.exchange));
    acc_request.set_api_version(mea.api);
    return acc_request;
  }

 private:
  coin2::app::Driver* driver_;
  py::object pipe_incoming_;
  py::object pipe_outgoing_;

  // telemetry
  std::unique_ptr<StrategyLogger> strat_logger_ = {};
  std::unique_ptr<TelemetryLogger> telemetry_logger_ = {};
  const coin2::feed::mgr::impl::FeedSubsystemManager* feed_mgr_ = nullptr;
  bool enable_telemetry_ = false;
};  // namespace impl

}  // namespace impl

using impl::PickleForwarder;
}  // namespace coin2::exchange::driver
